package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class Demo1Api {
    private HConnection connection;

    private HBaseAdmin hBaseAdmin;

    private TableName tableName = TableName.valueOf("shujia:student");

    @Before
    public void init() {
        // 1、创建连接

        Configuration configuration = new Configuration();

        /*
         * 不需要执行hmaster和regionserver的地址m  hclient会自动去zookeeper中获取，
         * 由于获取到的是一个域名，所以本地必须配置hosts
         *
         */

        //设置zookeeper地址
        configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        try {

            //创建连接  对数据增删改查使用
            connection = HConnectionManager.createConnection(configuration);


            /*
             * 创建修改删除表使用
             *
             */
            hBaseAdmin = new HBaseAdmin(connection);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void createTable() {

        //创建表的描述对象，指定表名
        HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf("shujia:student1"));

        //指定列簇
        HColumnDescriptor info = new HColumnDescriptor("info");
        info.setInMemory(true);

        hTableDescriptor.addFamily(info);

        try {

            //创建表使用hBaseAdmin
            hBaseAdmin.createTable(hTableDescriptor);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void dropTable() {
        try {


            if (hBaseAdmin.tableExists(TableName.valueOf("shujia:student"))) {
                //先禁用再删除
                hBaseAdmin.disableTable(TableName.valueOf("shujia:student"));

                hBaseAdmin.deleteTable(TableName.valueOf("shujia:student"));

            } else {
                System.out.println("表不存在");
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void put() {
        try {

            //获取表实例对象   ,  通过这个实例可以对表的数据进行增删改查
            HTableInterface table = connection.getTable(tableName);

            //创建put
            Put put = new Put("row1".getBytes());

            //增加列和数据
            put.add("info".getBytes(), "name".getBytes(), "zhsngsan".getBytes());

            put.add("info".getBytes(), "age".getBytes(), "23".getBytes());

            //插入数据
            table.put(put);

        } catch (IOException e) {
            e.printStackTrace();
        }

    }


    @Test
    public void puAll() {

        try {

            ArrayList<Put> puts = new ArrayList<>();

            BufferedReader bufferedReader = new BufferedReader(new FileReader("data/students.txt"));

            String line;

            while ((line = bufferedReader.readLine()) != null) {
                String[] split = line.split(",");
                String id = split[0];
                String name = split[1];
                String age = split[2];
                String gender = split[3];
                String clazz = split[4];


                //构建put以学号作为rowkey
                Put put = new Put(id.getBytes());

                //增加列
                put.add("info".getBytes(), "name".getBytes(), name.getBytes());
                put.add("info".getBytes(), "age".getBytes(), age.getBytes());
                put.add("info".getBytes(), "gender".getBytes(), gender.getBytes());
                put.add("info".getBytes(), "clazz".getBytes(), clazz.getBytes());


                puts.add(put);

            }


            HTableInterface table = connection.getTable(tableName);

            //批量插入数据
            table.put(puts);


        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Test
    public void get() {
        try {


            long start = System.currentTimeMillis();


            HTableInterface table = connection.getTable(tableName);

            //创建get
            Get get = new Get("1500100007".getBytes());

            //不设置列默认返回所有列
            //get.addColumn("info".getBytes(), "name".getBytes());
            //get.addFamily("info".getBytes());


            //查询数据
            Result result = table.get(get);

            //hbase保存的是二进制数据
            byte[] value = result.getValue("info".getBytes(), "name".getBytes());

            //将字节数组转换成字符串
            //String name = new String(value, "utf-8");
            String name = Bytes.toString(value);


            String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
            String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
            String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));


            System.out.println(name + "\t" + age + '\t' + gender + '\t' + clazz);


            long end = System.currentTimeMillis();


            System.out.println("查询用时:" + (end - start));

        } catch (IOException e) {
            e.printStackTrace();
        }

    }


    @Test
    public void get1() {
        try {
            HTableInterface table = connection.getTable(tableName);

            //创建get
            Get get = new Get("1500100007".getBytes());

            //拆线呢数据
            Result result = table.get(get);


            //一个cell代表一列
            List<Cell> cells = result.listCells();

            for (Cell cell : cells) {

                //获取列值
                String value = Bytes.toString(CellUtil.cloneValue(cell));

                //获取列民
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));

                //获取列簇
                String family = Bytes.toString(CellUtil.cloneFamily(cell));

                //获取rowkey
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));


                System.out.println(rowkey + "\t" + family + "\t" + qualifier + "\t" + value);

            }

        } catch (IOException e) {
            e.printStackTrace();
        }

    }


    @Test
    public void scan() {
        try {
            HTableInterface table = connection.getTable(tableName);

            Scan scan = new Scan();
            //如果什么条件都不加,那就是全表扫描

            //扫描表
            ResultScanner scanner = table.getScanner(scan);


            Result result;

            /// 每一个result代表一行
            while ((result = scanner.next()) != null) {

                //获取rowkey
                byte[] row = result.getRow();
                String id = Bytes.toString(row);

                //解析result
                String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
                String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
                String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
                String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));


                System.out.println(id + "\t" + name + "\t" + age + '\t' + gender + '\t' + clazz);


            }


        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @After
    public void close() {
        try {
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

        try {
            hBaseAdmin.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


}
